package br.com.felinelayer.common;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 */
public final class Communication implements Runnable {

    private DataOutputStream dataOutputStream = null;

    protected synchronized Socket getSocket() {
        return communicationListener.getSocket();
    }
    private ICommunicationListener communicationListener;

    public Communication(ICommunicationListener communicationListener) {
        setCommunicationListener(communicationListener);
    }

    public void disconnect() {
        try {
            getSocket().close();
        } catch (IOException ex) {
            Logger.getLogger(Communication.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public void setCommunicationListener(ICommunicationListener communicationListener) {
        try {
            this.communicationListener = communicationListener;
            this.dataOutputStream = new DataOutputStream(getSocket().getOutputStream());
            System.out.println("Communication: waiting message from other side");
            Thread thread = new Thread(this);
            thread.start();
        } catch (IOException ex) {
            Logger.getLogger(Communication.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
    private int _inquireId = 0;

//    private static final SimpleDateFormat idder = new SimpleDateFormat( "mmssSSS" );
    private synchronized int incInquireId() {
        int r = ++_inquireId;
        return r;
    }
    private Map<Integer, Object> waitingInquireLock = new HashMap<>();

    private synchronized void putInquireLock(Integer id, Object th) {
        waitingInquireLock.put(id, th);
    }

    private synchronized Object getInquireLock(Integer id) {
        return waitingInquireLock.get(id);
    }
    private Map<Integer, Message> waitingInquireResponse = new HashMap<>();

    private synchronized Message getInquireResponse(Integer id) {
        return waitingInquireResponse.get(id);
    }

    private synchronized void putInquireResponse(Integer id, Message message) {
        waitingInquireResponse.put(id, message);
    }

    private synchronized void removeInquireResponse(Integer id) {
        waitingInquireResponse.remove(id);
    }
    private HashMap<Thread, ArrayList<Message>> _batch = new HashMap<>();

    private synchronized ArrayList<Message> batch() {
        ArrayList<Message> result = _batch.get(Thread.currentThread());
        if (result == null) {
            result = new ArrayList<>();
            _batch.put(Thread.currentThread(), result);
        }
        return result;
    }
    private boolean inBatch = false;

    public Communication beginBatch() {
        inBatch = true;
        return this;
    }

    public boolean inBatch() {
        return inBatch;
    }

    private Message getMessageFromBatch(String name, int componentId) {
        for (Message c : batch()) {
            if (c.getIntByName(Protocol.COMPONENT_ID) != componentId) {
                continue;
            }
            String n = c.getName();
            if (n.equals(name)) {
                return c;
            }
        }
        return null;
    }

    private Message getMessageFromBatch(String[] names, int componentId) {
        for (Message c : batch()) {
            if (c.getIntByName(Protocol.COMPONENT_ID) != componentId) {
                continue;
            }
            String[] ns = c.getNames();
            if (Arrays.equals(ns, names)) {
                return c;
            }
        }
        return null;
    }

    private synchronized Communication addBatch(Message message) {
        ArrayList<Message> batch = batch();
        if (message.getName().equals(Protocol.MSG_NAME_COMPONENT_PROPERTY_CHANGE)) {
            int componentId = message.getIntByName(Protocol.COMPONENT_ID);
            // Check if exist message in que to some component and property then rewrite
            Message m = getMessageFromBatch(message.getNames(), componentId);
            if (m != null) {
                batch.remove(m);
            } else {
                // Check if exists message in queue to some component
                m = getMessageFromBatch(Protocol.MSG_NAME_CREATE_COMPONENT, componentId);
                if (m == null) {
                    m = getMessageFromBatch(Protocol.MSG_NAME_COMPONENT_PROPERTY_CHANGE,
                            componentId);
                }
                // If found message with property of some component then acumulates
                if (m != null) {
                    m.insertValuesFrom(message);
                    return this;
                }
            }
        }
        batch.add(message);
        return this;
    }

    public synchronized Communication sendMessage(Message message) {
        if (inBatch) {
            addBatch(message);
            return this;
        }
        message = markTimeSent(message);
        sendMessage(message.toString());
        return this;
    }

    private Message markTimeSent(Message message) {
        if (message.isPropertyExists(Protocol.MSG_TIME_SENT)) {
            message.setValueByName(Protocol.MSG_TIME_SOURCE_SENT, message.getLongByName(
                    Protocol.MSG_TIME_SENT));
        }
        message.setValueByName(Protocol.MSG_TIME_SENT, System.currentTimeMillis());
        return message;
    }

    public synchronized Communication endBatch() {
        ArrayList<Message> batch = batch();
        inBatch = false;
        if (batch.size() > 0) {
            markTimeSent(batch.get(0));
            String fullString;
            if (batch().size() == 1) {
                fullString = batch.get(0).toString();
            } else {
                fullString = Message.arrayToMessage(batch).toString();
            }
            sendMessage(fullString);
            batch.clear();
        }
        return this;
    }
    private static final SimpleDateFormat formatter = new SimpleDateFormat("hh:mm:ss:SSS");
    private long last = 0;

    public String now() {
        Date n = new Date();
        long t = n.getTime();
        long diff = 0;
        if (last != 0) {
            diff = t - last;
        }
        last = t;
        return formatter.format(n) + " " + diff;
    }

    //TODO: criar a thread somente para enviar? q cria um RunnAble que contém a mensagem, 
    // daí a thread singleton consome o runnable
    public synchronized void sendMessage(String message) {
        try {
            // Create output data stream
            if (communicationListener.getSocket().isClosed()) {
                System.out.println(prefix() + "impossível enviar comando, cliente fechado");
                return;
            }
            // Print a line to output data stream
            //System.out.println( "PROTOCOL " + now() + " Sent : " + message );
            FelLogger.log(prefix() + 
                    //now() + 
                    " Sent : " + message);
            // 
            writeText(dataOutputStream, message);
            //
        } catch (IOException ex) {
            Logger.getLogger(Communication.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
    
    public boolean isServer = false;
    
    private String prefix() {
        if (isServer)
            return "PROTOCOL SERVER: ";
        else
            return "PROTOCOL CLIENT: ";
    }

    public void finishMessageReceived(Message message) {
        // If need send inquire reply (confirmation)
        if (message.isPropertyExists(Protocol.INQUIRING_ID)) {
            int id = message.getIntByName(Protocol.INQUIRING_ID);
            message.deleteValue(Protocol.INQUIRING_ID);
            message.setValueByName(Protocol.INQUIRED_ID, id);
            sendMessage(message);
        }
    }

    private String readText(DataInputStream dataInputStream) throws IOException {
        return dataInputStream.readUTF();
    }

    private void writeText(DataOutputStream dataOutputStream, String message) throws IOException {
        dataOutputStream.writeUTF(message);
        dataOutputStream.flush();
    }

    public Message inquire(Message message) {
        if (inBatch) {
            addBatch(message);
            return message;
        }
        try {
            Inquire inq = createInquire();
            message.setValueByName(Protocol.INQUIRING_ID, inq.id.toString());
            sendMessage(message);
            synchronized (inq.getObject()) {
                inq.getObject().wait();
            }
            message = getInquireResponse(inq.getId());
            removeInquireResponse(inq.getId());
            return message;
        } catch (InterruptedException ex) {
            Logger.getLogger(Communication.class.getName()).log(Level.SEVERE, null, ex);
            return null;
        }
    }

    private Inquire createInquire() {
        Inquire inq = new Inquire();
        int inquireId = incInquireId();
        Object thread = new Object(); //Thread
        inq.setObject(thread).setId(new Integer(inquireId));
        putInquireLock(inquireId, thread);
        return inq;
    }

    class Inquire {

        private Object thread;
        private Integer id;

        public Inquire setId(Integer id) {
            this.id = id;
            return this;
        }

        public Integer getId() {
            return this.id;
        }

        public Inquire setObject(Object object) {
            this.thread = object;
            return this;
        }

        public Object getObject() {
            return thread;
        }
    }

    @Override
    public void run() {
        try {
            DataInputStream input = new DataInputStream(getSocket().getInputStream());
            // Loop to listen from other side
            while (true) {
                if ((input == null) || (getSocket() == null) || (getSocket().isClosed())) {
                    System.out.println(prefix() + "peer desconnected!");
                    break;
                }
                String text = readText(input);
                if ((text == null) || text.isEmpty()) {
                    continue;
                }
                // Convert text message do protocol object
                Message message = new Message().setFromString(text);

                FelLogger.log(prefix() + 
                        //"latency " + getLantency(message) + " echo "
                        //+ getEchoLantency(message) + 
                        " Received : " + text);

                // Start thread to fire event from message received 
                new FireMessageReceived().fireMessageReceived(this, communicationListener, message);
            }
            if (input != null) {
                input.close();
            }
            dataOutputStream.close();
        } catch (Exception ex) {
            String m = ex.getMessage();
            if (((m != null) && (m.indexOf("socket closed") >= 0)) || ((m != null)
                    && (m.indexOf("Connection reset") >= 0))
                    || (ex instanceof java.io.EOFException)) {
                System.out.println(prefix() + "peer desconectou!");
            } else {
                Logger.getLogger(Communication.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    private long getLantency(Message message) {
        long diff = -1;
        if (message.isPropertyExists(Protocol.MSG_TIME_SENT)) {
            diff = System.currentTimeMillis() - message.getLongByName(
                    Protocol.MSG_TIME_SENT);
        }
        return diff;
    }

    private long getEchoLantency(Message message) {
        long diff = -1;
        if (message.isPropertyExists(Protocol.MSG_TIME_SOURCE_SENT)) {
            diff = System.currentTimeMillis() - message.getLongByName(
                    Protocol.MSG_TIME_SOURCE_SENT);
        }
        return diff;
    }

    // run -> new FireMessageReceived
    class FireMessageReceived extends Thread {

        private Communication communication;
        private ICommunicationListener communicationListener;
        private Message message;

        public void fireMessageReceived(Communication communication,
                ICommunicationListener communicationListener, Message message) {
            this.communication = communication;
            this.communicationListener = communicationListener;
            this.message = message;
            start();
        }

        @Override
        public void run() {
            communicationListener.processMessageReceived(message);
            // If must wait inquire response
            communication.wakeUpInquiringThread(message);
        }
    }

    /**
     * responseInquire
     *
     * @param message
     */
    private void wakeUpInquiringThread(Message message) {
        if (!message.isPropertyExists(Protocol.INQUIRED_ID)) {
            return;
        }
        //
        String sId = message.getValueByName(Protocol.INQUIRED_ID);
        if (sId.isEmpty()) {
            return;
        }
        Integer id = new Integer(sId);
        Object o = getInquireLock(id);
        if ((o != null) && (o instanceof Object)) {
            Object th = (Object) o; // Thread
            putInquireResponse(id, message);
            //System.out.println( "thread " + id + " acordando ..." );
            synchronized (th) {
                th.notify();
            }
            //System.out.println( "thread " + id + " acordei!" );
        } else {
            System.out.println(prefix() + "thread " + id + " não encontrada! mensagem:" + message.toString());
        }
    }
}